package com.prb.register.register.impl;

import com.prb.common.config.RpcConfig;
import com.prb.common.statistical.TimeMetrics;
import com.prb.register.register.RpcRegistryHandler;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.prefs.NodeChangeListener;

/**
 * 基于zookeeper实现的注册中心<br />
 * zookeeper保存路径：/lg-edu-rpc/com.lagou.edu.rpc.api.UserService/provider/127.0.0.1:8890
 */
public class ZookeeperRegistryHandler implements RpcRegistryHandler {

    private static String spilt = "/";

    private List<String> serverList = new CopyOnWriteArrayList<>();

    private ZkClient client;

    public ZkClient getClient() {
        return client;
    }

    private static ScheduledExecutorService EXPORT_WORKER = Executors.newScheduledThreadPool(1);


    //初始化 zk 客户端
    public ZookeeperRegistryHandler(String url) {

        client = new ZkClient(url);

        //客户端，开启定时统计调用时间功能
        if(RpcConfig.serverSide.equals("client")){

            EXPORT_WORKER.scheduleWithFixedDelay(()-> {
                    TimeMetrics.metricsMap.forEach((key, value) -> {
                        System.out.println("before : " + TimeMetrics.metricsMap);
                        TimeMetrics.metricsMap.remove(key);
                        String path = spilt + key + spilt + value.getIp() + ":" + value.getPort();
                        System.out.println("定时上报对象" + value);
                        if(client.exists(path)){
                            try {
                                client.writeData(path , value.getEnd().toString().getBytes("UTF-8"));
                            } catch (UnsupportedEncodingException e) {
                                e.printStackTrace();
                            }
                        }
                        System.out.println("after : " + TimeMetrics.metricsMap);
                    });

                },2, 3 ,TimeUnit.SECONDS);

        }

    }

    //服务端需要注册服务
    @Override
    public boolean registry(String service, String ip, int port) {

        //分布式并行注册，可能会出现数据安全问题，出现只是概率问题，需要加分布式锁，暂时不添加
        if(!client.exists(spilt + service)){
            //如果第一个服务注册，则需要添加 调用信息 和 主机信息
            client.createPersistent(spilt + service);
        }


        client.createEphemeral(spilt + service + spilt + ip + ":" + port);

        return true;
    }

    //客户端需要发现服务
    @Override
    public List<String> discovery(String service) {

        //如果服务列表 为空， 则需要从 zk 中 获取， 反之直接返回，后续服务列表变更，通过 zk 的监听机制进行更新
        //List<String> list<ip:port>
        if(serverList.isEmpty()){
            serverList = client.getChildren(spilt + service);
        }
        return serverList;
    }

    private void registryWatch(final String service, final String path) {

    }

    @Override
    public void destroy() {

        if (null != client) {
            client.close();
        }
    }

    @Override
    public void addListener(NodeChangeListener listener) {

    }
}
